{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "## aiohttp\n",
    "## github地址：\n",
    "## https://github.com/aio-libs/aiohttp\n",
    "## aiohttp官方文档：\n",
    "## https://docs.aiohttp.org/en/stable/\n",
    "\n",
    "## sanic\n",
    "## github地址：\n",
    "## https://github.com/channelcat/sanic\n",
    "\n",
    "## aiomysql\n",
    "## github地址：\n",
    "## https://github.com/aio-libs/aiomysql\n",
    "## aiomysql 也可以用来集成SQLAlchemy\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import asyncio\n",
    "import re\n",
    "\n",
    "import aiohttp\n",
    "import aiomysql\n",
    "from pyquery import PyQuery\n",
    "\n",
    "## 爬虫启停标志\n",
    "stopping = False\n",
    "start_url = \"http://www.jobbole.com/\"\n",
    "## 共享队列\n",
    "waitting_urls = []\n",
    "## 设置过滤器（大数据参考布隆过滤器）\n",
    "seen_urls = set()\n",
    "\n",
    "async def fetch(url, session):\n",
    "    try:\n",
    "        async with session.get(url) as resp:\n",
    "            print(\"rl status: {}\".format(resp.status))\n",
    "            if resp.status in [200, 201]:\n",
    "                data = await resp.text()\n",
    "                return data\n",
    "    except Exception as e:\n",
    "        print(e)\n",
    "\n",
    "## 解析html中的url\n",
    "def extract_urls(html):\n",
    "    urls = []\n",
    "    pq = PyQuery(html)\n",
    "    for link in pq.items(\"a\"):\n",
    "        url = link.attr(\"href\")\n",
    "        if url and url.startswith(\"http\") and url not in seen_urls:\n",
    "            urls.append(url)\n",
    "            waitting_urls.append(url)\n",
    "    return urls\n",
    "\n",
    "## 获取入口url的html\n",
    "async def init_urls(url, session):\n",
    "    html = await fetch(url, session)\n",
    "    seen_urls.add(url)\n",
    "    extract_urls(html)\n",
    "\n",
    "## 获取文章详情，并且解析入库\n",
    "async def article_handler(url, session, pool):\n",
    "    ## 获取页面\n",
    "    html = await fetch(url, session)\n",
    "    ## 放入爬去队列\n",
    "    extract_urls(html)\n",
    "    ## 添加已爬取url\n",
    "    seen_urls.add(url)\n",
    "\n",
    "    ## 获取字段\n",
    "    pq = PyQuery(html)\n",
    "    title = pq(\"title\").text()\n",
    "\n",
    "    ## 连接数据库\n",
    "    async with pool.acquire() as conn:\n",
    "        async with conn.curse() as cur:\n",
    "            await cur.execute(\"SELECT 42;\")\n",
    "            ## 写入数据库\n",
    "            insert_sql = \"insert into article_test(title) values ('{}')\".format(title)\n",
    "            await cur.execute(insert_sql)\n",
    "\n",
    "\n",
    "## 消费者，等待数据产生\n",
    "async def consumer(pool):\n",
    "    ## 生产全局的session，并传递给后续的函数\n",
    "    async with aiohttp.ClientSession() as session:\n",
    "        while not stopping:\n",
    "            if len(waitting_urls) == 0:\n",
    "                await asyncio.sleep(0.5)\n",
    "                continue\n",
    "            url = waitting_urls.pop()\n",
    "            print(\"start get url: {}\".format(url))\n",
    "            if re.match(\"http://.*?jobbole.com/\\d+/\", url):\n",
    "                ## 如果匹配成功，且没有抓取过\n",
    "                if url not in seen_urls:\n",
    "                    asyncio.ensure_future(article_handler(url, session, pool))\n",
    "                    ## 调试\n",
    "                    await asyncio.sleep(0.5)\n",
    "            else:\n",
    "                ## 如果没有匹配成功\n",
    "                if url not in seen_urls:\n",
    "                    asyncio.ensure_future(init_urls(url, session))\n",
    "\n",
    "async def main(loop):\n",
    "    ## 等待MySQL建立连接\n",
    "    pool = await aiomysql.create_pool(host='127.0.0.1', port=3306,\n",
    "                                      user='root', password='123456',\n",
    "                                      db='mysql', loop=loop,\n",
    "                                      charset='utf8', autocommit=True)\n",
    "    ## 启动抓取页面\n",
    "    # asyncio.ensure_future(init_urls(start_url))\n",
    "    # asyncio.ensure_future(consumer(pool))\n",
    "\n",
    "    async with aiohttp.ClientSession() as session:\n",
    "        html = await fetch(start_url, session)\n",
    "        seen_urls.add(start_url)\n",
    "        extract_urls(html)\n",
    "\n",
    "    asyncio.ensure_future(consumer(pool))\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    loop = asyncio.get_event_loop()\n",
    "    asyncio.ensure_future(main(loop))\n",
    "    loop.run_forever()\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
